package com.i72.logextend.appender;

import ch.qos.logback.classic.PatternLayout;
import ch.qos.logback.classic.spi.LoggingEvent;
import ch.qos.logback.core.UnsynchronizedAppenderBase;
import ch.qos.logback.core.encoder.Encoder;
import ch.qos.logback.core.spi.DeferredProcessingAware;
import ch.qos.logback.core.status.ErrorStatus;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.*;
import org.elasticsearch.common.xcontent.XContentType;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;


public class RollingESAppender<E> extends UnsynchronizedAppenderBase<E> {


    //以下是配置



    /**
     * 编码组件
     */
    private Encoder<E> encoder;

    /**
     * 日志缓冲条数，到达或超过这个临界值将后写入
     */
    private int bufferLogCount = 200;

    /**
     * 日志缓冲时间，到达或超过这个时间间隔将写入 ms
     */
    private long bufferTimeMs = 5000l;

    /**
     * es 集群地址
     */
    private String esCluster = "";


    private String indexPattern = "";

    //以下是内部变量
    private final Object lock = new Object();

    private volatile boolean working;

    private List<byte[]> events = new LinkedList<>();

    //private List<byte[]> errorEvents = new LinkedList<>();

    private RestHighLevelClient client = null;

    private ReentrantLock reentrantLock = new ReentrantLock();

    private ReentrantLock errorLogsLock = new ReentrantLock();

    private PatternLayout patternLayout = null;

    private ExecutorService executorService = null;

    private int corePoolSize = 2;

    private int maximumPoolSize = 10;

    private int queueCapacity = 1000;

    private static List<byte[]> errorEvents = new LinkedList<>();

    //private AtomicInteger num = new AtomicInteger(0);
    //private AtomicInteger logNum = new AtomicInteger(0);



    @Override
    public void start() {
        super.start();
        initES();
        initPatternLayout();

        executorService = new ThreadPoolExecutor(corePoolSize,maximumPoolSize,30, TimeUnit.SECONDS,new LinkedBlockingDeque<>(queueCapacity));


    }

    @Override
    public void stop() {
        super.stop();
        disposeES();
    }

    @Override
    public void doAppend(E eventObject) {
        super.doAppend(eventObject);
    }


    @Override
    protected void append(E eventObject) {
        if (!isStarted()) {
            return;
        }

        if (client!=null && !working) {

            if (!working) {

                //打开一个无限循环的线程，用来监控日志数量
                working = true;
                Thread thread = new Thread(new MonitorRunnable());
                thread.start();
            }
        }


        if(client!=null) {
            try {
                // this step avoids LBCLASSIC-139
                if (eventObject instanceof DeferredProcessingAware) {
                    ((DeferredProcessingAware) eventObject).prepareForDeferredProcessing();
                }
                // the synchronization prevents the OutputStream from being closed while we
                // are writing. It also prevents multiple threads from entering the same
                // converter. Converters assume that they are in a synchronized block.
                // lock.lock();

                //LoggingEventCompositeJsonEncoder encoder = (LoggingEventCompositeJsonEncoder)this.encoder;

                byte[] byteArray = this.encoder.encode(eventObject);

                reentrantLock.lock();
                events.add(byteArray);
                reentrantLock.unlock();


            } catch (Exception ioe) {
                // as soon as an exception occurs, move to non-started state
                // and add a single ErrorStatus to the SM.
                this.started = false;
                addStatus(new ErrorStatus("IO failure in appender", this, ioe));
            }
        }

    }

    public void initES(){
        if(client==null){
            synchronized (lock){
                if(client==null){

                    if(esCluster!=null && esCluster.length()>0) {

                        String[] esNode = esCluster.split(";");

                        if (esNode.length > 0){


                            Node [] nodeArr = new Node[esNode.length];
                            for(int i=0;i<esNode.length;i++){
                                Node n = new Node(HttpHost.create(esNode[i]));
                                nodeArr[i] = n;
                            }

                            RestClientBuilder builder = RestClient.builder(nodeArr);
                            client = new RestHighLevelClient(builder);

                            if(client==null){
                                addWarn("ES连接出现问题，请检查");
                                System.out.println("ES连接出现问题，请检查");

                            }
                        }
                    }
                }
            }
        }
    }

    public void initPatternLayout(){
        PatternLayout patternLayout = new PatternLayout();
        patternLayout.setContext(context);
        patternLayout.setPattern(this.getIndexPattern());
        patternLayout.setOutputPatternAsHeader(false);
        patternLayout.start();
        this.patternLayout = patternLayout;
    }

    public void disposeES(){
        if(client!=null){
            try {
                client.close();
                client = null;
            }catch (Exception e){

            }
        }
    }

    //监控线程
    class MonitorRunnable implements Runnable{
        @Override
        public void run() {
            long prevMs = System.currentTimeMillis();
            while (true) {

                long currentMs = System.currentTimeMillis();

                if(events.size()>=bufferLogCount || currentMs-prevMs>=bufferTimeMs) {
                    if(events.size()>0) {
                        reentrantLock.lock();
                        List<byte[]> eventsCopy = events;
                        events = new LinkedList<>();
                        reentrantLock.unlock();
                        try {
                            //用来日志发送
                            executorService.execute(new BulkRunnable(eventsCopy));
                            //num.incrementAndGet();
                            //logNum.getAndAdd(eventsCopy.size());
                            //System.out.println("提交任务:"+eventsCopy.size()+"条。提交任务:"+num.toString()+"次。当前共:"+logNum.toString()+"条");

                        } catch (Exception e) {
                            System.out.println("线程池出现问题:" + e.getMessage());
                        }
                    }

                    prevMs = System.currentTimeMillis();
                }

                try {
                    Thread.sleep(10l);
                }catch (Exception e){
                    System.out.println("监控出现问题:"+e.getMessage());
                }
            }
        }
    }

    //日志上送逻辑，允许在线程池内
    class BulkRunnable implements Runnable{

        private List logs = null;

        public BulkRunnable(List<byte[]> logs) {
            this.logs = logs;
        }

        @Override
        public void run() {

            try {

                if (errorEvents.size() > 0) {
                    errorLogsLock.lock();
                    logs.addAll(errorEvents);
                    errorEvents.clear();
                    errorLogsLock.unlock();
                }

                if(logs.size()>0) {
                    bulk(logs);
                }

            } catch (Exception e) {

            }
        }

        public boolean bulk(List<byte[]> logs){


            boolean result = true;

            BulkRequest bulkRequest = new BulkRequest();

            //FileNamePattern namePattern = new FileNamePattern(indexPattern,this.con);
            LoggingEvent event = new LoggingEvent();
            event.setTimeStamp(new Date().getTime());
            String index =  patternLayout.doLayout(event);

            for(byte[] bytes:logs){

                bulkRequest.add(new IndexRequest(index,"logs").
                        source(bytes, XContentType.JSON));
            }

            //long t1 = System.currentTimeMillis();

            try {

                client.bulk(bulkRequest, RequestOptions.DEFAULT);

                //System.out.println("es索引:"+index+"日志耗时:"+new Long(System.currentTimeMillis()-t1).toString()+"。插入:"+logs.size()+"条日志");
            }catch (Exception e){
                addWarn("创建es日志出错");
                System.out.println("创建es日志出错");
                //产生新集合插入，可能会有重复。消费时去重即可
                errorLogsLock.lock();
                errorEvents.addAll(logs);
                errorLogsLock.unlock();
                result = false;
            }finally {
                logs.clear();
                logs = null;
            }
            return result;
        }
    }

    public int getBufferLogCount() {
        return bufferLogCount;
    }

    public void setBufferLogCount(int bufferLogCount) {
        this.bufferLogCount = bufferLogCount;
    }

    public long getBufferTimeMs() {
        return bufferTimeMs;
    }

    public void setBufferTimeMs(long bufferTimeMs) {
        this.bufferTimeMs = bufferTimeMs;
    }

    public String getEsCluster() {
        return esCluster;
    }

    public void setEsCluster(String esCluster) {
        this.esCluster = esCluster;
    }

    public Encoder<E> getEncoder() {
        return encoder;
    }

    public void setEncoder(Encoder<E> encoder) {
        this.encoder = encoder;
    }

    public String getIndexPattern() {
        return indexPattern;
    }

    public void setIndexPattern(String indexPattern) {
        this.indexPattern = indexPattern;
    }

    public int getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    public int getMaximumPoolSize() {
        return maximumPoolSize;
    }

    public void setMaximumPoolSize(int maximumPoolSize) {
        this.maximumPoolSize = maximumPoolSize;
    }

    public int getQueueCapacity() {
        return queueCapacity;
    }

    public void setQueueCapacity(int queueCapacity) {
        this.queueCapacity = queueCapacity;
    }
}
